"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 数据库系统
Case Name   : 1主1备1级联,备节点事务提交方式为local
Create At   : 2021/02/25
Owner       : opentestcase026
Description :
    1.初始化环境
    2.将备节点1切换为级联备
    3.设置事务同步方式为local,synchronous_standby_names为空
    4.配置所有节点备份延迟为2min
    5.查询集群同步方式
    6.创建表并插入数据
    7.等待5s查询备机
    8.等待2min查询备机
    9.插入数据,60s后再插入数据
    10.等待5s查询备机
    11.等待70s查询备机
    12.等待70s查询备机
    13.删除数据
    14.等待2min修改延迟参数修改为0
    15.备机查询
    16.清理环境
Expect      :
    1.成功
    2.切换成功
    3.设置成功
    4.设置成功
    5.级联备为异步,备节点为异步
    6.创建并插入成功
    7.查询不到
    8.数据同步
    9.数据插入成功
    10.备节点未同步
    11.备节点同步3但未同步到4
    12.备节点可查到所有数据
    13.删除成功
    14.配置成功
    15.备节点同步
    16.成功
History     :
    modified:2021/03/11 by opentestcase001:增加时间打印,辅助定位
    Modified by opentestcase026 2021/3/26:增加查询时间
    modified:2021/05/14 by opentestcase001:数据库状态显示格式变化,修改脚本适配
    modified:2022/07/13 by opentestcase004:优化代码格式&研发代码变更,修改脚本适配新代
    码
"""
import unittest
import time
import os
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant
from testcase.utils.CommonSH import CommonSH

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == Primary_SH.get_node_num(),
                 '需主备环境,若为单机环境则不执行')
class RecoveryDelay(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("----this is setup----")
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.db_primary_user_node = Node(node='PrimaryDbUser')
        self.com = Common()
        self.comshpri = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.tb_name = 'tb_case0011'
        self.rootnodelist = ['Standby1Root', 'Standby2Root']

    def test_recovery_delay(self):
            text = '----step1:初始化环境 expect:成功----'
            self.log.info(text)
            self.log.info('----1.1获取节点数目----')
            self.nodelist = ['Standby1DbUser', 'Standby2DbUser']
            result = self.comshpri.get_db_cluster_status('detail')
            self.log.info(result)
            self.node_num = result.count('Standby Normal') + 1
            self.comshsta = []
            self.log.info(self.node_num)
            self.assertEqual(self.node_num, 3, '执行失败' + text)
            for i in range(int(self.node_num) - 1):
                self.comshsta.append(CommonSH(self.nodelist[i]))
            self.log.info('----1.2安装ntp服务----')
            install_cmd = "yum install -y ntp && " \
                          "systemctl enable ntpd && systemctl start ntpd"
            for i in range(int(self.node_num) - 1):
                db_standby_node = Node(node=self.rootnodelist[i])
                result = db_standby_node.sh(install_cmd).result()
                self.log.info(install_cmd)
                self.log.info(result)
            self.log.info('----1.3同步集群时间----')
            current = self.db_primary_user_node.sh(
                "date \"+%m/%d/%Y %H:%M:%S\"").result()
            self.log.info(current)
            datecmd = f'date -s "{current}"'
            for i in range(int(self.node_num) - 1):
                db_standby_node = Node(node=self.rootnodelist[i])
                result = db_standby_node.sh(datecmd).result()
                self.log.info(datecmd)
                self.log.info(result)

            text = '----step2:将备节点1切换为级联备 expect:切换成功----'
            self.log.info(text)
            self.log.info('----2.1将备节点1切换为级联备,初始化环境----')
            result = self.comshpri.get_db_cluster_status('detail')
            self.log.info(result)
            result = self.comshsta[0].build_standby('-M cascade_standby')
            self.assertIn(self.constant.REBUILD_SUCCESS_MSG, result)
            result = self.comshsta[0].exec_refresh_conf()
            self.assertTrue(result, '执行失败' + text)
            self.log.info('----2.2查询数据库状态----')
            result = self.comshpri.get_db_cluster_status('detail')
            self.log.info(result)
            if 'Cascade Normal' not in result and 'Stanby Normal' not in \
                    result:
                result = self.comshpri.restart_db_cluster()
                self.log.info(result)
                self.assertTrue(result, '执行失败' + text)

            text = '----step3:设置synchronous_commit=local,synchronous_standby' \
                   '_names为空 expect:设置成功----'
            self.log.info(text)
            self.result = self.com.show_param('synchronous_standby_names')
            if self.result != '':
                re = self.comshpri.execute_gsguc(
                    'reload', self.constant.GSGUC_SUCCESS_MSG,
                    "synchronous_standby_names=''")
                self.log.info(re)
                self.assertTrue(re, '执行失败' + text)
            self.result1 = self.com.show_param('synchronous_commit')
            if self.result1 != 'local':
                re = self.comshpri.execute_gsguc(
                    'reload', self.constant.GSGUC_SUCCESS_MSG,
                    'synchronous_commit=local')
                self.assertTrue(re, '执行失败' + text)

            text = '----step4:设置recovery_min_apply_delay=2min ' \
                   'expect:设置成功----'
            self.log.info(text)
            self.result2 = self.com.show_param('recovery_min_apply_delay')
            if self.result2 != '2min':
                re = self.comshpri.execute_gsguc(
                    'reload', self.constant.GSGUC_SUCCESS_MSG,
                    'recovery_min_apply_delay=2min')
                self.assertTrue(re, '执行失败' + text)
            self.log.info('--------查询参数--------')
            result = self.comshpri.execut_db_sql('show synchronous_commit;')
            self.log.info(result)
            self.assertIn('local', result, '执行失败' + text)
            result = self.comshpri.execut_db_sql(
                'show recovery_min_apply_delay;')
            self.log.info(result)
            self.assertIn('2min', result, '执行失败' + text)

            text = '----step5:查询集群同步方式 expect:级联备为异步,' \
                   '备节点为异步----'
            self.log.info(text)
            result = self.comshpri.get_db_cluster_status('detail')
            self.log.info(result)
            sql = "select * from pg_stat_replication;"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertIn('Async', result, '执行失败' + text)
            result = self.comshsta[1].execut_db_sql(sql)
            self.log.info(result)
            self.assertIn('Async', result, '执行失败' + text)
            self.log.info('--------等待主备一致------------')
            result = self.comshsta[1].check_data_consistency()
            self.assertTrue(result)
            for i in range(90):
                result = self.comshpri.check_cascade_standby_consistency()
                if result:
                    break
                time.sleep(20)
            self.assertTrue(result)

            text = '----step6:创建表,并插入数据 expect:创建并插入成功----'
            self.log.info(text)
            sql = f"drop table if exists {self.tb_name};" \
                f"create table {self.tb_name}(i int, s char(10));" \
                f"insert into {self.tb_name} values(5,'test');"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result,
                          '执行失败' + text)
            self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败'
                          + text)

            text = '----step7:等待5秒,备机查询 expect:查询不到----'
            self.log.info(text)
            time.sleep(5)
            sql = f"select * from {self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                self.assertIn(self.constant.NOT_EXIST, result, '执行失败' +
                              text)

            text = '----step8:等待120s,查询备机 expect:数据同步----'
            self.log.info(text)
            time.sleep(120)
            sql = f"select sysdate;select * from {self.tb_name};"
            for exec_time in range(5):
                flag = 0
                for i in range(int(self.node_num) - 1):
                    result = self.comshsta[i].execut_db_sql(sql)
                    self.log.info(result)
                    if '5 | test' in result:
                        flag += 1
                if flag > 1:
                    break
                time.sleep(10)
                self.assertEqual(2, flag, '执行失败' + text)

            text = '----step9:插入数据,60s后再插入数据 expect:数据插入成功----'
            self.log.info(text)
            sql = f"insert into {self.tb_name} values(3, 'test');"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertIn(self.constant.INSERT_SUCCESS_MSG, result,
                          '执行失败' + text)
            time.sleep(60)
            sql = f"insert into {self.tb_name} values(4, 'test');"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败'
                          + text)

            text = '----step10:等待5s,查询备机 expect:备节点未同步----'
            self.log.info(text)
            time.sleep(5)
            sql = f"select sysdate;select * from {self.tb_name};" \
                f"select count(*) from {self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                self.assertNotIn('3 | test', result)
                self.assertNotIn('4 | test', result)
                self.assertIn('1', result)
                self.assertIn('5 | test', result)

            text = '----step11:等待70s查询备机 expect:备节点同步3但未同' \
                   '步到4----'
            self.log.info(text)
            time.sleep(70)
            sql = f"select sysdate;select * from {self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                judge_flg = ('3 | test' in result) and \
                    ('5 | test' in result) and ('2 rows' in result) and \
                    ('4 | test' not in result)
                self.assertTrue(judge_flg, '执行失败' + text)

            text = '----step12:等待70s查询备机 expect:备节点可查到所有数据----'
            self.log.info(text)
            time.sleep(70)
            sql = f"select sysdate;select * from {self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                judge_flg = ('3 | test' in result) and \
                    ('5 | test' in result) and ('3 rows' in result) \
                    and ('4 | test' in result)
                self.assertTrue(judge_flg, '执行失败' + text)

            text = '----step13:删除数据 expect:删除成功----'
            self.log.info(text)
            sql = f"delete from {self.tb_name};"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertIn('DELETE', result, '执行失败' + text)

            text = '----step14:设置recovery_min_apply_delay=0 ' \
                   'expect:配置成功----'
            self.log.info(text)
            time.sleep(120)
            result = self.comshpri.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                'recovery_min_apply_delay=0')
            self.assertTrue(result, '执行失败' + text)

            text = '----step15:备机查询expect:备节点同步----'
            self.log.info(text)
            sql = f"select * from {self.tb_name};"
            flg = 0
            for exe_i in range(30):
                flg = 0
                for i in range(int(self.node_num) - 1):
                    result = self.comshsta[i].execut_db_sql(sql)
                    self.log.info(result)
                    if '(0 rows)' in result:
                        flg = flg + 1
                if 2 == flg:
                    break
                time.sleep(10)
            self.assertEqual(2, flg, '执行失败' + text)

    def tearDown(self):
        text = '----step16:清理环境 expect:成功----'
        self.log.info(text)
        self.log.info('-----删除表----')
        sql = f"drop table if exists {self.tb_name};"
        result1 = self.comshpri.execut_db_sql(sql)
        self.log.info(result1)

        self.log.info('-----还原集群----')
        if self.node_num > 2:
            self.comshsta[0].build_standby('-M standby')
            result2 = self.comshsta[0].exec_refresh_conf()
            self.log.info(result2)
        result3 = self.comshpri.get_db_cluster_status('detail')
        self.log.info(result3)

        self.log.info('----------还原参数-------')
        cmd = f"alter system set synchronous_standby_names to '{self.result}'"
        re1 = self.comshpri.execut_db_sql(cmd)
        self.log.info(re1)
        re2 = self.comshpri.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'synchronous_commit={self.result1}')
        self.log.info(re2)
        re3 = self.comshpri.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'recovery_min_apply_delay={self.result2}')
        self.log.info(re3)

        self.assertIn('ALTER', re1, '执行失败' + text)
        self.assertTrue(re2, '执行失败' + text)
        self.assertTrue(re3, '执行失败' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1, '执行失败' +
                      text)
        self.assertEqual(result3.count('Standby Normal'), 2, '执行失败' + text)
        self.log.info(f'----{os.path.basename(__file__)} end----')
